package com.wpre.vampiremq.mqserver.datacenter;

import com.wpre.vampiremq.common.BinaryTool;
import com.wpre.vampiremq.common.MQException;
import com.wpre.vampiremq.mqserver.core.MSGQueue;
import com.wpre.vampiremq.mqserver.core.Message;

import java.io.*;
import java.util.LinkedList;
import java.util.Scanner;

/**
 * Created with IntelliJ IDEA.
 * Description:管理硬盘上的消息文件
 * User: LianBao
 * Date: 2025-01-14
 * Time: 0:28
 */
public class MessageFileManager {

    public void init() {
        //初始化方法,目前不需要
    }

    //定义内部类,表示该队列的统计信息(因为只有当前的类使用到Stat)
    static public class Stat {
        public int totalCount;// 总消息数量
        public int validCount;// 有效消息数量
    }

    //获取指定队列对应的消息文件所在路径
    private String getQueueDir(String queueName) {
        return "./data/" + queueName;
    }

    //获取消息数据文件的路径
    private String getQueueDataPath(String queueName) {
        return getQueueDir(queueName) + "/queue_data.txt";
    }

    //获取队列的消息统计文件的路径
    private String getQueueStatPath(String queueName) {
        return getQueueDir(queueName) + "/queue_stat.txt";
    }

    /**
     * 读取消息统计文件
     *
     * @param queueName
     * @return
     */
    private Stat readStat(String queueName) {
        Stat stat = new Stat();
        try (InputStream inputStream = new FileInputStream(getQueueStatPath(queueName))) {
            //消息的统计文件是txt文件,所以可以直接使用Scanner读取
            Scanner scanner = new Scanner(inputStream);
            stat.totalCount = scanner.nextInt();//总消息数
            stat.validCount = scanner.nextInt();//合法消息数
            return stat;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }


    /**
     * 写入消息统计文件
     *
     * @param queueName
     * @param stat
     */
    private void writeStat(String queueName, Stat stat) {
        try (OutputStream outputStream = new FileOutputStream(getQueueStatPath(queueName))) {
            PrintWriter printWriter = new PrintWriter(outputStream);
            printWriter.write(stat.totalCount + "\t" + stat.validCount);
            printWriter.flush();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    /**
     * 创建队列对应的文件和目录
     *
     * @param queueName
     * @throws IOException
     */
    public void createQueueFiles(String queueName) throws IOException {
        // 1. 创建队列对应的目录
        File baseDir = new File(getQueueDir(queueName));
        if (!baseDir.exists()) {
            //如果不存在就创建
            boolean ok = baseDir.mkdirs();
            if (!ok) {
                throw new IOException("创建队列对应的目录失败");
            }
        }
        // 2.创建队列数据文件
        File queueDataFile = new File(getQueueDataPath(queueName));
        if (!queueDataFile.exists()) {
            //如果不存在就创建
            boolean newFile = queueDataFile.createNewFile();
            if (!newFile) {
                throw new IOException("创建队列数据文件失败");
            }
        }
        // 3.创建消息统计文件
        File queueStatFile = new File(getQueueStatPath(queueName));
        if (!queueStatFile.exists()) {
            boolean newFile = queueStatFile.createNewFile();
            if (!newFile) {
                throw new IOException("创建消息统计文件失败");
            }
        }
        //4. 给消息统计文件赋初始值
        Stat stat = new Stat();
        stat.validCount = 0;
        stat.totalCount = 0;
        //写入stat
        writeStat(queueName, stat);
    }

    /**
     * 删除队列对应的目录和文件
     *
     * @param queueName
     */
    public void deleteQueueFile(String queueName) throws IOException {
        // 1.先删除数据文件
        File queueDataFile = new File(getQueueDataPath(queueName));
        boolean delete = queueDataFile.delete();

        //再删除Stat文件
        File queueStatFile = new File(getQueueStatPath(queueName));
        boolean delete1 = queueStatFile.delete();

        //再删除目录
        File queueDir = new File(getQueueDir(queueName));
        boolean delete2 = queueDir.delete();

        if (!delete || !delete1 || !delete2) {
            //删除失败
            throw new IOException("删除队列对应的目录和文件失败");
        }
    }

    // 检查队列的目录和文件是否存在
    // 生产者给BrokerServer 生产消息时,可能会持久化,就要写入文件,要先检测目录和文件是否存在
    public boolean checkFilesExist(String queueName) {
        //判断队列的数据文件和统计文件是否都存在
        File queueDataFile = new File(getQueueDataPath(queueName));
        if (!queueDataFile.exists()) {
            return false;
        }

        File queueStatFile = new File(getQueueStatPath(queueName));
        if (!queueStatFile.exists()) {
            return false;
        }

        return true;
    }

    /**
     * 把新的消息,放到对应的队列中
     *
     * @param queue   要写入的队列
     * @param message 要写的消息
     */
    public void sendMessage(MSGQueue queue, Message message) throws IOException {
        // 1.检查要写入的队列对应的文件是否存在
        if (!checkFilesExist(queue.getName())) {
            throw new MQException("要写入的队列文件不存在");
        }

        // 2.写入文件前,先把Message序列化

        byte[] messageBinary = BinaryTool.toBytes(message);
        synchronized (queue) {
            // 3.写入文件前,获取当前队列数据文件的长度,从而计算出Message对象的offsetBeg 和 offsetEnd
            //(只有在写文件的时候,才能知道offsetBeg 和 offsetEnd)
            // 新的 Message 数据写入到队列数据文件的末尾,此时的Message对象的offsetBeg 就是当前文件长度+4 ,offsetEnd就是当前文件长度+4+Message自身长度
            File queueDataFile = new File(getQueueDataPath(queue.getName()));
            //获取文件长度queueDataFile.length();
            //设置offsetBeg和offsetEnd
            message.setOffsetBeg(queueDataFile.length() + 4);//当前文件长度+4
            message.setOffsetEnd(queueDataFile.length() + 4 + messageBinary.length);//当前文件长度+4+Message自身长度

            // 4. 写入消息到数据文件中
            try (OutputStream outputStream = new FileOutputStream(queueDataFile, true)) {
                //参数设置true表示追加写入
                try (DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                    // 1. 先写消息的长度(4个字节)
                    //messageBinary.length 是一个int 类型, 需要把它的4个字节依次写入到文件中
                    //使用OutputStream 的write方法其实写入的是byte类型的,所以我们选择DataOutputStream
                    dataOutputStream.writeInt(messageBinary.length);
                    // 2. 写消息本体
                    dataOutputStream.write(messageBinary);

                }
            }

            // 5.更新消息统计文件
            Stat stat = readStat(queue.getName());
            stat.totalCount += 1;
            stat.validCount += 1;
            writeStat(queue.getName(), stat);
        }
    }

    //删除消息,逻辑删除,把isValid设置为0即可
    public void deleteMessage(MSGQueue queue, Message message) throws IOException, ClassNotFoundException {

        synchronized (queue) {
            try (RandomAccessFile randomAccessFile = new RandomAccessFile(getQueueDataPath(queue.getName()), "rw")) {
                //先把文件中这段数据读取出来,还原成Message对象
                byte[] bufferSrc = new byte[(int) (message.getOffsetEnd() - message.getOffsetBeg())];
                //移动光标
                randomAccessFile.seek(message.getOffsetBeg());
                //把数据读取到bufferSrc数组中
                randomAccessFile.read(bufferSrc);
                //转换成Message对象
                Message diskMessage = (Message) BinaryTool.fromBytes(bufferSrc);
                //把isValid设置为0即可
                diskMessage.setIsValid((byte) 0x0);

                //把数据重新写回文件
                byte[] bufferDest = BinaryTool.toBytes(diskMessage);

                //重新移动光标到offsetBeg(光标会随着读写进行变化)
                randomAccessFile.seek(message.getOffsetBeg());
                //更新后的消息写回文件
                randomAccessFile.write(bufferDest);
            }
            //更新统计文件中的有效消息个数
            Stat stat = readStat(queue.getName());
            if (stat.validCount > 0) {
                stat.validCount -= 1;
            }
            writeStat(queue.getName(), stat);
        }

    }

    /**
     * 从文件中读取出所有的消息,加载到内存中(保存在链表中),这个方法在程序启动时调用
     * 参数是queueName,而不是MSGQueue对象,是因为这个方法不需要加锁,不涉及多线程操作(因为是程序启动的时候调用的)
     *
     * @param queueName
     * @return LinkedList方便头插头删
     */
    public LinkedList<Message> loadAllMessageFromQueue(String queueName) throws IOException, ClassNotFoundException {
        LinkedList<Message> messages = new LinkedList<>();
        try (InputStream inputStream = new FileInputStream(getQueueDataPath(queueName))) {
            try (DataInputStream dataInputStream = new DataInputStream(inputStream)) {
                long currentOffset = 0;//记录当前文件光标位置
                while (true) {
                    //循环读取
                    //1.读取当前消息的长度
                    int messageLen = dataInputStream.readInt();//读取4个字节
                    //readInt读到文件末尾会抛出异常,只需要捕获异常

                    //2.根据这个长度,读取消息数据
                    byte[] buffer = new byte[messageLen];
                    int readLen = dataInputStream.read(buffer);
                    if (readLen != messageLen) {
                        //如果实际读取的消息长度不等于当前消息长度
                        throw new MQException("loadAllMessageFromQueue 文件格式错误");
                    }
                    //3. 把读取的二进制数据转换成Message对象
                    Message message = (Message) BinaryTool.fromBytes(buffer);

                    //4.判断是否是无效数据
                    if (message.getIsValid() != 0x1) {
                        //跳过无效数据
                        currentOffset += (4 + messageLen);//更新光标
                        continue;
                    }
                    //5. 设置Message对象的offset属性
                    message.setOffsetBeg(currentOffset + 4);
                    message.setOffsetEnd(currentOffset + 4 + messageLen);
                    currentOffset += (4 + messageLen);//更新光标

                    //6. 把有效数据加入到链表中
                    messages.add(message);

                }

            }
        } catch (EOFException e) {

        }
        return messages;
    }

    //检查当前队列是否需要进行垃圾回收
    public boolean checkGC(String queueName) {
        Stat stat = readStat(queueName);
        if (stat.totalCount > 1999 &&
                (double) stat.validCount / (double) stat.totalCount < 0.3) {
            return true;
        }
        return false;
    }

    private String getQueueDataNewPath(String queueName) {
        return getQueueDir(queueName) + "/queue_data_new.txt";
    }

    //垃圾回收
    public void GC(MSGQueue queue) throws IOException, ClassNotFoundException {
        //把有效数据拷贝到新文件中,删除旧文件,再把新的文件重命名,更新消息统计文件
        synchronized (queue) {
            //创建新文件
            File newDataFile = new File(getQueueDataNewPath(queue.getName()));
            if (newDataFile.exists()) {
                //如果这个文件存在,说明上次GC出问题
                throw new MQException("GC出现问题");
            }
            boolean ok = newDataFile.createNewFile();
            if (!ok) {
                throw new IOException("GC时创建文件失败");
            }
            //2. 从旧的文件读取所有的有效消息对象
            LinkedList<Message> messages = loadAllMessageFromQueue(queue.getName());

            //3. 把有效消息写入新的文件中
            try (OutputStream outputStream = new FileOutputStream(newDataFile)) {
                try (DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                    for (Message message : messages) {
                        byte[] bytes = BinaryTool.toBytes(message);
                        //先写4个字节的长度
                        dataOutputStream.writeInt(bytes.length);
                        dataOutputStream.write(bytes);
                    }
                }

            }
            //4.删除旧文件,重命名新文件
            File oldDataFile = new File(getQueueDataPath(queue.getName()));
            boolean delete = oldDataFile.delete();
            if (!delete) {
                throw new MQException("删除旧数据文件失败");
            }
            //重命名
            boolean b = newDataFile.renameTo(oldDataFile);
            if (!b) {
                throw new MQException("重命名失败");
            }
            //更新消息统计文件
            Stat stat = readStat(queue.getName());
            //垃圾回收之后,全部都是有效消息
            stat.totalCount = messages.size();
            stat.validCount = messages.size();

        }

    }


}
